package reader.file;

import writer.es.TraceDataBean;
import writer.es.TaskQueue;

import java.io.*;
import java.util.Date;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;

// 文件读线程
// 支持对写入完成的文件, 使用线程池进行消费。
// 支持第一次全量读文件夹下的文静, 并监听文件新增并消费
public class FileReader implements Runnable {
    private CountDownLatch latch;
    private String fileName = "";
    private BlockingQueue<Map<String, File>> queue;
    private File s = null;
    private FileInputStream fis = null;
    private BufferedReader br = null;
    private InputStreamReader isr = null;

    public FileReader(BlockingQueue<Map<String, File>> queue, CountDownLatch latch)
    {
        this.queue = queue;
        this.latch = latch;
    }

    //@Override
    public void run(){
        while (true) {
            try {
                long startTime = System.currentTimeMillis();
                Map<String, File> map = queue.take();
                int count = 0;
                Set<String> set = map.keySet();
                for (Iterator<String> iter = set.iterator(); iter.hasNext();) {
                    fileName = iter.next().toString();
                    s = map.get(fileName);
                    try {
                        // TODO: 2023/2/5 这里如果不休眠,会抛出找不到文件异常,因为文件被其他进程占用 
                        try {
                            Thread.sleep(5000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        fis = new FileInputStream(s);
                        isr = new InputStreamReader(fis, "utf-8");
                        br = new BufferedReader(isr);
                        String line = "";
                        while ((line = br.readLine()) != null) {
                            // TODO: 2023/2/5 数据的映射
                            String[] fields = line.split("\\|");
                            TraceDataBean dataBean = new TraceDataBean();
                            dataBean.setTel(fields[0]);
                            dataBean.setLng(Double.parseDouble(fields[5]));
                            dataBean.setLat(Double.parseDouble(fields[6]));
                            dataBean.setEventTime(new Date(new Long(fields[1])));
                            TaskQueue.getQueue().add(dataBean);
                            count++;
                        }
                        long endTime = System.currentTimeMillis();
                        System.out.println(Thread.currentThread().getName() + " 线程读文件 ：" + fileName + " 完毕" + " ,条数为 " + count + " ,时间为 " + (endTime - startTime)/1000 + "s");
                        latch.countDown();
                    } catch (FileNotFoundException e) {
                        e.printStackTrace();
                    } catch (IOException e) {
                        e.printStackTrace();
                    } finally {
                        try {
                            fis.close();
                            isr.close();
                            br.close();
                            //消费完成并删除文件
                            s.delete();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }
                // TODO: 2023/2/5 如果使用虚拟内存mmp是否更快？
//                MappedByteBuffer buffer=null;
//                for (Iterator<String> iter = set.iterator(); iter.hasNext();) {
//                    fileName = iter.next().toString();
//                    s = map.get(fileName);
//                    long len = s.length();
//                    byte[] ds = new byte[(int) len];
//                    try {
//                        buffer=new RandomAccessFile(s,"rw").getChannel().map(FileChannel.MapMode.READ_WRITE, 0, len);
//                    } catch (IOException e) {
//                        e.printStackTrace();
//                    }
//                    int n = 0;
//                    for(int i=0;i<len;i++){
//                        ds[i] = buffer.get(i);
//                        n=0x000000ff&ds[i];
//                        count+=n;
//
//                    }
//                }


            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}